from pyspark.sql import SparkSession,Window
from pyspark.sql.functions import *
from pyspark.sql.types import *

spark = SparkSession \
    .builder \
    .master('local') \
    .appName('HelloSpark')\
    .getOrCreate()


file_path=r"C:\Users\Administrator"

# user_df=spark.read.json(file_path+"\yelp_academic_dataset_user.json")
# business_df=spark.read.json(file_path+"\yelp_academic_dataset_business.json")
review_df=spark.read.json(file_path+"\yelp_academic_dataset_review.json")
# checkin_df=spark.read.json(file_path+"\yelp_academic_dataset_checkin.json")


# 一、商户分析
# 1.1 找出美国最常见商户（前20）
# print("1.1 找出美国最常见商户（前20）:")
# business_df.select("name") \
#     .groupBy("name") \
#     .agg(count('name').alias('count')) \
#     .orderBy(col('count').desc()) \
#     .show(20)

# 1.2 找出美国商户最多的前10个城市
# print("1.2 找出美国商户最多的前10个城市")
# business_df.select('city') \
#     .groupBy('city') \
#     .agg(count('city').alias('count')) \
#     .orderBy(col('count').desc())\
#     .show(10)

# 1.3 找出美国商户最多的前5个州
# print("1.3 找出美国商户最多的前5个州")
# business_df.select('state') \
#     .groupBy('state') \
#     .agg(count('state').alias('count')) \
#     .orderBy(col('count').desc()) \
#     .show(5)

# 1.4 找出美国最常见商户，并显示平均评分（前20）
# print("1.4 找出美国最常见商户，并显示平均评分（前20）")
# business_df.select('name','stars') \
#     .groupBy('name') \
#     .agg(count('name').alias('count'),avg('stars').alias('avg_stars')) \
#     .orderBy(col('count').desc()) \
#     .show(20)

# 1.5 统计评分最高的城市（前10）
# print("1.5 统计评分最高的城市（前10）")
# business_df.select('city','stars') \
#     .groupBy('city') \
#     .agg(avg('stars').alias('avg_stars'))\
#     .orderBy(col('avg_stars').desc()) \
#     .show(10)

# 1.6 统计category的数量
# print("1.6 统计category的数量")
# business_df.select('categories')\
#     .withColumn('category',explode(split(col('categories'),','))) \
#     .agg(countDistinct('category')) \
#     .show()

# 1.7
# print(" 1.7 统计最多的category及数量（前10）")
# business_df.select('categories')\
#     .withColumn('category',explode(split(col('categories'),', '))) \
#     .groupBy('category') \
#     .agg(count("*").alias('count')) \
#     .orderBy(col('count').desc()) \
#     .show(10)

# 1.8 收获五星评论最多的商户（前20）
# print("1.8 收获五星评论最多的商户（前20）")
# business_df.select("name",'stars') \
#     .where('stars == 5') \
#     .groupBy('name') \
#     .agg(count('name').alias('count')) \
#     .orderBy(col('count').desc()) \
#     .show(20)

# 1.9 统计不同类型（中国菜、美式、墨西哥）的餐厅类型及数量
# print("1.9统计不同类型（中国菜、美式、墨西哥）的餐厅类型及数量")
# business_df_Restaurants=business_df.select('categories').filter(col('categories').like('%Restaurants%'))
# business_df_Restaurants\
#     .withColumn('category',explode(split(col('categories'),', ')))  \
#     .where("category like '%Chinese%' or category like '%American%' or category like '%Mexican%'") \
#     .groupBy('category') \
#     .agg(count("category").alias('count')) \
#     .show()

# 1.10 统计不同类型（中国菜、美式、墨西哥）的餐厅的评论数量
# print("10 统计不同类型（中国菜、美式、墨西哥）的餐厅的评论数量")
# business_df_Restaurants=business_df.select('categories','review_count').filter(col('categories').like('%Restaurants%'))
# business_df_Restaurants\
#     .withColumn('category',explode(split(col('categories'),', ')))  \
#     .where("category like '%Chinese%' or category like '%American%' or category like '%Mexican%'") \
#     .groupBy('category') \
#     .agg(sum('review_count').alias('sum_count')) \
#     .show()

# 1.11 统计不同类型（中国菜、美式、墨西哥）的餐厅的评分分布
# print("1.11 统计不同类型（中国菜、美式、墨西哥）的餐厅的评分分布")
# business_df_Restaurants=business_df.select('categories','review_count').filter(col('categories').like('%Restaurants%'))
# business_df_Restaurants\
#     .withColumn('category',explode(split(col('categories'),', ')))  \
#     .where("category like '%Chinese%' or category like '%American%' or category like '%Mexican%'") \
#     .groupBy('category','review_count') \
#     .agg(sum('review_count').alias('sum_count')) \
#     .show()


# 二、用户分析
# 2.1 分析每年加入的用户数量
# user_df.select('yelping_since')\
#     .withColumn('year',year('yelping_since'))\
#     .groupBy('year') \
#     .agg(count('year').alias('count')) \
#     .orderBy('year') \
#     .show(truncate=False)
#
#  2.2 统计评论达人（review_count）
# user_df.select('name','review_count')  \
#     .orderBy(col("review_count").desc())\
#     .show(truncate=False)

# 2.3 统计人气最高的用户（fans）
# user_df.select('name','fans') \
#     .orderBy(col('fans').desc()) \
#     .show(truncate=False)

# 2.4 统计每年优质用户、普通用户比例
# year_df=user_df.withColumn('year',year('yelping_since')).select('year').distinct().orderBy(col('year').asc())
# year_df_2=user_df.withColumn('year',year('yelping_since')).select('year').distinct().orderBy(col('year').asc())


# 得到每一年的总人数
# def getUserNumber(Pyear):
#     print("hello2")
#     # print(Pyear)
#     result=(year_df_2.where(f'year <= {Pyear}').agg(count("*")).collect()[0])[0]
#     return result
#
# getNumber = udf(getUserNumber,IntegerType())
# print("hello")
#
# year_df.withColumn('totalNumber',getNumber(col('year'))).show()
# print("hello5")
#
# result=(user_df.withColumn('year',year('yelping_since')).where(f'year <= 2013').agg(count("*")).collect()[0])[0]
# print(type(result))

# 三、评分分析
# 3.1 统计每年的评论数
review_df.select('date')\
    .withColumn('year',year('date')) \
    .groupBy('year') \
    .agg(count('year').alias('count')) \
    .orderBy(col('year').desc()) \
    .show()

# 3.2 统计评分周（周一~周天）次数统计






